package com.hzh.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo15mapPartitions {
  def main(args: Array[String]): Unit = {
    /**
     * 建立连接
     */

    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo13Sort")
    val sc = new SparkContext(conf)

    val linesRDD: RDD[String] = sc.textFile("data/words")

    println(linesRDD.getNumPartitions)

    /**
     * mapPartitions:一次处理一个分区的数据，将一个分区的数据一个一个的传递给后面的函数
     *
     * 迭代器中是一个分区的数据
     *
     */
    linesRDD.mapPartitions((iter: Iterator[String]) => {
      iter.flatMap(_.split(","))
    })
      .foreach(println)

    /**
     * mapPartitionsWithIndex:对一个分区编号进行处理
     */

    linesRDD.mapPartitionsWithIndex {
      case ((index: Int, iter: Iterator[String])) =>
        println(s"当前分区编号：$index")
        iter

    }.foreach(println)


  }

}
